package com.ferry.transport.netty.provider;

import com.ferry.common.util.Pair;
import com.ferry.common.util.SystemClock;
import com.ferry.common.util.reflect.Reflects;
import com.ferry.protocol.StatusCode;
import com.ferry.protocol.payload.RequestPayload;
import com.ferry.protocol.payload.ResponsePayload;
import com.ferry.serial.api.Serializer;
import com.ferry.serial.api.SerializerFactory;
import com.ferry.transport.netty.concurrent.CloseableExecutor;
import com.ferry.transport.netty.concurrent.ExecutorFactory;
import com.ferry.transport.netty.concurrent.ThreadPoolExecutorFactory;
import com.ferry.transport.netty.server.DefaultServiceProviderContainer;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by daiyong
 */
public class DefaultProviderProcessor implements ProviderProcessor {

	private static final Logger logger = LoggerFactory.getLogger(DefaultProviderProcessor.class);

	private DefaultServiceProviderContainer defaultServiceProviderContainer;

	private CloseableExecutor executor;

	public DefaultProviderProcessor() {
		this.executor = new ThreadPoolExecutorFactory().newExecutor(ExecutorFactory.Target.PROVIDER, "ferry-provider");
	}

	public DefaultProviderProcessor(DefaultServiceProviderContainer defaultServiceProviderContainer) {
		this();
		this.defaultServiceProviderContainer = defaultServiceProviderContainer;
	}

	@Override
	public void handleRequest(Channel channel, RequestPayload requestPayload) throws Exception {
		Request request = new Request();
		request.setRequestPayload(requestPayload);
		executor.execute(new MessageTask(this, channel, request));
	}

	@Override
	public void handleException(Channel channel, RequestPayload request, StatusCode statusCode, Throwable cause) {
		logger.warn("handleException, remoteAddress:{}", channel.remoteAddress(), cause);

		//处理异常
		RespWrapper respWrapper = new RespWrapper();
		// 截断cause, 避免客户端无法找到cause类型而无法序列化
		respWrapper.setError(cause);

		//获取序列化器
		Serializer serializer = SerializerFactory.getSerializer(request.serializerCode());

		ResponsePayload response = new ResponsePayload(request.invokeId());
		response.status(statusCode.getCode());
		byte[] bytes = serializer.writeObject(respWrapper);
		response.bytes(request.serializerCode(), bytes);

		ChannelFuture channelFuture = channel.writeAndFlush(response);
		channelFuture.addListener(ChannelFutureListener.CLOSE);  //关闭长连接
	}

	@Override
	public void shutdown() {
		if (null != executor) {
			executor.shutdown();
		}
	}

	@Override
	public ServiceWrapper lookup(String uniKey) {
		return defaultServiceProviderContainer.lookupService(uniKey);
	}

	public void setDefaultServiceProviderContainer(DefaultServiceProviderContainer defaultServiceProviderContainer) {
		this.defaultServiceProviderContainer = defaultServiceProviderContainer;
	}
}

class MessageTask implements Runnable {

	private static final Logger logger = LoggerFactory.getLogger(MessageTask.class);

	private ProviderProcessor providerProcessor;
	private Channel channel;
	private Request request;

	public MessageTask(ProviderProcessor providerProcessor, Channel channel, Request request) {
		this.providerProcessor = providerProcessor;
		this.channel = channel;
		this.request = request;
	}

	@Override
	public void run() {

		//序列化为 MessageWrapper
		MessageWrapper msg;
		try {

			RequestPayload requestPayload = request.getRequestPayload();

			byte serializerCode = requestPayload.serializerCode();
			Serializer serializer = SerializerFactory.getSerializer(serializerCode);

			byte[] bytes = requestPayload.bytes();
			msg = serializer.readObject(bytes, MessageWrapper.class);
			requestPayload.clear();

			request.setMessageWrapper(msg);


			//寻找服务
			ServiceWrapper serviceWrapper = providerProcessor.lookup(msg.getServiceMetaData().directoryString());
			if (null == serviceWrapper) {
				providerProcessor.handleException(channel, request.getRequestPayload(), StatusCode.SERVICE_NOT_FOUND, new RuntimeException("server not found"));
				return;
			}

			process(serviceWrapper);
		} catch (Throwable t) {
			providerProcessor.handleException(channel, request.getRequestPayload(), StatusCode.SERVER_ERROR, t);
			return;
		}

	}

	private void process(ServiceWrapper serviceWrapper) throws NoSuchMethodException {
		Object provider = serviceWrapper.getServiceProvider();
		MessageWrapper msg = request.getMessageWrapper();
		String methodName = msg.getMethodName();
		Object[] args = msg.getArgs();

		Class<?>[] expectCauseTypes = null;
		List<Pair<Class<?>[], Class<?>[]>> methodExtension = serviceWrapper.getMethodExtension(methodName);
		if (methodExtension == null) {
			throw new NoSuchMethodException(methodName);
		}

		// 根据JLS方法调用的静态分派规则查找最匹配的方法parameterTypes
		Pair<Class<?>[], Class<?>[]> bestMatch = Reflects.findMatchingParameterTypesExt(methodExtension, args);
		Class<?>[] parameterTypes = bestMatch.getFirst();
		expectCauseTypes = bestMatch.getSecond();

		Object resultObj = Reflects.fastInvoke(provider, methodName, parameterTypes, args);

		//写回数据
		RespWrapper result = new RespWrapper();
		result.setResp(resultObj);
		byte s_code = request.getRequestPayload().serializerCode();
		Serializer serializer = SerializerFactory.getSerializer(s_code);

		ResponsePayload responsePayload = new ResponsePayload(request.getRequestPayload().invokeId());

		byte[] bytes = serializer.writeObject(result);
		responsePayload.bytes(s_code, bytes);
		responsePayload.status(StatusCode.OK.getCode());


		ChannelFuture channelFuture = channel.writeAndFlush(responsePayload);
		channelFuture.addListener(new ChannelFutureListener() {
			@Override
			public void operationComplete(ChannelFuture future) throws Exception {
				logger.info("write success");
			}
		});
	}
}
